package com.scpii.universal.concurrent;

import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 任务处理线程池
 * 
 * @author chenxin
 * 
 */
public class TaskPool<T> {

	private static final Logger log = LoggerFactory.getLogger(TaskPool.class);

	private ExecutorService exec;

	private CompletionService<T> completionService;

	/**
	 * 新建任务处理线程池
	 * 
	 * @param procThreadNum
	 *            线程池线程处理数
	 */
	public TaskPool(int procThreadNum) {
		exec = Executors.newFixedThreadPool(procThreadNum);
		completionService = new ExecutorCompletionService<T>(exec);
	}

	/**
	 * 提交任务
	 * 
	 * @param task
	 *            任务
	 */
	public void submitTask(Callable<T> task) {
		completionService.submit(task);
	}

	/**
	 * 获取执行的FUTURE模式
	 * 
	 * @return
	 */
	public CompletionService<T> getCompletionService() {
		return completionService;
	}

	/**
	 * 获取执行线程池
	 * 
	 * @return
	 */
	public ExecutorService getExec() {
		return exec;
	}

	/**
	 * 线程池中活动的线程数
	 * 
	 * @return
	 */
	public int getActiveCount() {
		ThreadPoolExecutor tpool = (ThreadPoolExecutor) exec;
		return tpool.getActiveCount();
	}

	/**
	 * 关闭线程池
	 */
	public void shutDown() {
		log.info("禁止新的任务提交");
		exec.shutdown();
		try {
			log.info("等待3分钟,中止现存的任务");
			if (!exec.awaitTermination(180, TimeUnit.SECONDS)) {
				exec.shutdownNow();
				if (!exec.awaitTermination(180, TimeUnit.SECONDS)) {
					log.info("线程池没有终止");
				}
			}
		} catch (InterruptedException ie) {
			exec.shutdownNow();
			Thread.currentThread().interrupt();
		}
	}

}